package com.xzzz.irda.tracker.server;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import com.xzzz.common.base.pojo.PageRes;
import com.xzzz.common.base.util.DateUtils;
import com.xzzz.expand.es.core.EsCurdManager;
import com.xzzz.irda.tracker.core.SpanNode;
import com.xzzz.irda.tracker.core.TrackerProperties;
import com.xzzz.irda.tracker.core.adapter.aspect.TrackerStart;
import com.xzzz.irda.tracker.core.repository.TrackerRepository;
import com.xzzz.irda.tracker.core.common.TrackerConstants;
import com.xzzz.irda.tracker.server.pojo.SpanNodeTermsRes;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Cancellable;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.*;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.BucketOrder;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.metrics.ParsedMax;
import org.elasticsearch.search.aggregations.metrics.ParsedMin;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.stream.Collectors;

/**
 * tracker 的各项信息由 elasticsearch 保存.
 * <p
 */
@Slf4j
@Service
public class TrackerMetricServiceByElasticSearch implements TrackerRepository, TrackerMetricService {

    @Autowired
    private EsCurdManager esManager;

    @Autowired
    private TrackerProperties properties;

    /**
     * 批量保存
     *
     * @param spanNodes span 集合
     */
    @Override
    public void saveBatch(List<SpanNode> spanNodes) {
        esManager.addBatch(TrackerConstants.ELASTIC_SEARCH_INDEX_NAME, spanNodes);
    }

    @Override
    @TrackerStart
    @Scheduled(cron = "0 0/1 * * * ?")
    public long expire() {
        long start = System.currentTimeMillis();
        Date today = DateUtil.date();
        // 删除3天前的
        Date expireDate = DateUtil.offsetDay(today, properties.getRepository().getElasticsearch().getExpireDay() * -1);
        String expireDateStr = DateUtil.format(expireDate, "yyyy-MM-dd HH:mm:ss");
        QueryBuilder query = QueryBuilders.rangeQuery("spanStart")
                .lt(expireDateStr).format("yyyy-MM-dd HH:mm:ss");

        ActionListener<BulkByScrollResponse> listener = new ActionListener<BulkByScrollResponse>() {
            @Override
            public void onResponse(BulkByScrollResponse bulkResponse) {
                log.info("[TRACKERS] 删除小于 [{}] SPAN 信息, 需要删除的条数: {}, 用时: {}",
                        expireDateStr, bulkResponse.getDeleted(), System.currentTimeMillis() - start);
            }

            @Override
            public void onFailure(Exception e) {
                log.warn("[TRACKERS] 删除失败: {}", e.getMessage());
            }
        };

        // BulkByScrollResponse deleteResponse = esCurdManager.deleteByQuery(TrackerConstants.ELASTIC_SEARCH_INDEX_NAME, query);
        // log.info("[TRACKERS] 删除小于 [{}] SPAN 信息, 需要删除的条数: {}, 用时: {}",
        //         expireDateStr, deleteResponse.getDeleted(), System.currentTimeMillis() - start);
        // return deleteResponse.getDeleted();

        Cancellable cancellable = esManager.deleteByQueryAsync(TrackerConstants.ELASTIC_SEARCH_INDEX_NAME, query, listener);
        return 0;
    }

    @Override
    public List<SpanNode> list(String traceId) {
        SearchSourceBuilder searchSource = new SearchSourceBuilder().size(100).sort("spanStart", SortOrder.ASC);

        BoolQueryBuilder bool = new BoolQueryBuilder();
        searchSource.query(bool);

        TermsQueryBuilder appNameTerm = QueryBuilders.termsQuery("traceId", traceId);
        bool.filter(appNameTerm);

        SearchResponse resp = esManager.search(TrackerConstants.ELASTIC_SEARCH_INDEX_NAME, searchSource);
        List<SpanNode> spanList = new ArrayList<>();

        SearchHits hits = resp.getHits();
        for (SearchHit hit : hits) {
            Map<String, Object> map = hit.getSourceAsMap();
            if (MapUtil.isNotEmpty(map)) {
                SpanNode spanNode = new SpanNode();
                spanNode.setAppName((String) map.get("appName"));
                spanNode.setSpanType((String) map.get("spanType"));
                spanNode.setTraceId((String) map.get("traceId"));
                spanNode.setSpanId((String) map.get("spanId"));
                spanNode.setSpanParentId((String) map.get("spanParentId"));
                spanNode.setSpanName((String) map.get("spanName"));
                spanNode.setSpanStart(Long.parseLong(map.get("spanStart").toString()));
                spanNode.setSpanEnd(Long.parseLong(map.get("spanEnd").toString()));
                spanNode.setSpanInterval(Long.parseLong(map.get("interval").toString()));
                spanNode.setRecords(new TreeMap<>((HashMap) map.get("records")));
                spanList.add(spanNode);
            }
        }
        return sortTreeSpans(fillParentWithOrphan(spanList));
    }

    /**
     * 查询Trace分页列表
     *
     * @param pageNum  当前页
     * @param pageSize 当前行数
     * @param appName  应用名称
     * @param traceId  追踪ID
     * @return
     */
    @Override
    public PageRes<SpanNode> pages(Integer pageNum, Integer pageSize, String appName, String traceId) {
        SearchSourceBuilder searchSource = new SearchSourceBuilder()
                .sort("spanStart", SortOrder.DESC)
                .timeout(TimeValue.timeValueSeconds(5)).from((pageNum - 1) * pageSize).size(pageSize);

        BoolQueryBuilder bool = new BoolQueryBuilder();
        searchSource.query(bool);

        if (StrUtil.isNotBlank(appName)) {
            TermQueryBuilder appNameTerm = QueryBuilders.termQuery("appName", appName);
            bool.filter(appNameTerm);
        }

        if (StrUtil.isNotBlank(traceId)) {
            TermQueryBuilder traceIdTerm = QueryBuilders.termQuery("traceId", traceId);
            bool.filter(traceIdTerm);
        }

        ExistsQueryBuilder notExistsSpanParentId = QueryBuilders.existsQuery("spanParentId");
        bool.mustNot(notExistsSpanParentId);

        SearchResponse resp = esManager.search(TrackerConstants.ELASTIC_SEARCH_INDEX_NAME, searchSource);

        PageRes<SpanNode> pageRes = new PageRes<>();
        List<SpanNode> pageData = new ArrayList<>();
        pageRes.setDatas(pageData);


        SearchHits hits = resp.getHits();
        pageRes.setTotal(hits.getTotalHits().value);
        pageRes.setPageSize(0);
        pageRes.setPageNum(pageNum);
        pageRes.setPageSize(pageSize);
        pageRes.setTotalPage((int) (pageRes.getTotal() / pageSize));

        for (SearchHit hit : hits) {
            Map<String, Object> map = hit.getSourceAsMap();
            if (MapUtil.isNotEmpty(map)) {
                SpanNode spanNode = new SpanNode();
                spanNode.setAppName((String) map.get("appName"));
                spanNode.setSpanType((String) map.get("spanType"));
                spanNode.setTraceId((String) map.get("traceId"));
                spanNode.setSpanName((String) map.get("spanName"));
                spanNode.setSpanStart(Long.parseLong(map.get("spanStart").toString()));
                spanNode.setSpanInterval(Long.parseLong(map.get("interval").toString()));
                pageData.add(spanNode);
            }
        }

        return pageRes;
    }

    /**
     * 根据 traceId 分组聚合查询
     *
     * @param pageSize 页数
     * @param appName  应用名称
     * @param traceId  追踪ID
     */
    @Override
    public List<SpanNodeTermsRes> listGroupByTraceId(Integer pageSize, String appName, String traceId) {
        SearchSourceBuilder searchSource = new SearchSourceBuilder().timeout(TimeValue.timeValueSeconds(5));
        BoolQueryBuilder bool = new BoolQueryBuilder();
        searchSource.query(bool);

        if (StrUtil.isNotBlank(appName)) {
            TermQueryBuilder appNameTerm = QueryBuilders.termQuery("appName", appName);
            bool.filter(appNameTerm);
        }

        if (StrUtil.isNotBlank(traceId)) {
            TermQueryBuilder traceIdTerm = QueryBuilders.termQuery("traceId", traceId);
            bool.filter(traceIdTerm);
        }

        AggregationBuilder termsByTraceIdAggs = AggregationBuilders.terms("termsByTraceId")
                .size(pageSize)
                .field("traceId")
                .order(BucketOrder.aggregation("minStart", false));

        AggregationBuilder minStartAggs = AggregationBuilders.min("minStart").field("spanStart").format("yyyy-MM-dd HH:mm:ss.SSS");
        AggregationBuilder maxEndAggs = AggregationBuilders.max("maxEnd").field("spanEnd").format("yyyy-MM-dd HH:mm:ss.SSS");
        termsByTraceIdAggs.subAggregation(minStartAggs).subAggregation(maxEndAggs);

        SearchResponse resp = esManager.search(TrackerConstants.ELASTIC_SEARCH_INDEX_NAME, searchSource.aggregation(termsByTraceIdAggs));

        List<SpanNodeTermsRes> result = new ArrayList<>();

        ParsedTerms terms = resp.getAggregations().get("termsByTraceId");
        List<? extends Terms.Bucket> buckets = terms.getBuckets();
        for (Terms.Bucket bucket : buckets) {
            SpanNodeTermsRes spanNode = new SpanNodeTermsRes();
            spanNode.setTraceId(bucket.getKeyAsString());
            spanNode.setNodeCount(bucket.getDocCount());

            ParsedMin minStart = bucket.getAggregations().get("minStart");
            spanNode.setSpanStart(DateUtils.toYMDHMS_SSS((long) minStart.getValue()));

            ParsedMax maxEnd = bucket.getAggregations().get("maxEnd");
            spanNode.setSpanInterval((long) (maxEnd.getValue() - minStart.getValue()));
            result.add(spanNode);
        }

        //

        List<String> traceIds = result.stream().map(SpanNodeTermsRes::getTraceId).collect(Collectors.toList());

        SearchSourceBuilder termsSearchSource = new SearchSourceBuilder()
                .timeout(TimeValue.timeValueSeconds(5)).from(0).size(traceIds.size())
                .fetchSource(new String[]{"spanName", "appName", "spanType", "traceId"}, null);

        BoolQueryBuilder bool1 = new BoolQueryBuilder();
        termsSearchSource.query(bool1);

        bool1.must(QueryBuilders.termQuery("spanParentId", TrackerConstants.ROOT_SPAN));
        bool1.must(QueryBuilders.termsQuery("traceId", traceIds.toArray()));

        Map<String, SpanNode> termHitMaps = new HashMap<>();

        SearchResponse termResp = esManager.search(TrackerConstants.ELASTIC_SEARCH_INDEX_NAME, termsSearchSource);
        for (SearchHit hit : termResp.getHits()) {
            Map<String, Object> map = hit.getSourceAsMap();
            if (MapUtil.isNotEmpty(map)) {
                SpanNode spanNode = new SpanNode();
                spanNode.setAppName((String) map.get("appName"));
                spanNode.setSpanName((String) map.get("spanName"));
                spanNode.setSpanType((String) map.get("spanType"));
                spanNode.setTraceId((String) map.get("traceId"));
                termHitMaps.put(spanNode.getTraceId(), spanNode);
            }
        }

        for (SpanNodeTermsRes spanNodeTermsRes : result) {
            SpanNode span = termHitMaps.get(spanNodeTermsRes.getTraceId());
            if (span == null) {
                continue;
            }

            spanNodeTermsRes.setAppName(span.getAppName());
            spanNodeTermsRes.setSpanName(span.getSpanName());
            spanNodeTermsRes.setSpanType(span.getSpanType());
        }

        return result;
    }

    /**
     * 将 span 信息持久化到 es 中
     *
     * @param spanNode span 信息集合
     * @return 是否保存成功
     */
    @Override
    public boolean save(List<SpanNode> spanNode) {
        try {
            esManager.addBatch(TrackerConstants.ELASTIC_SEARCH_INDEX_NAME, spanNode);
            return true;
        } catch (Exception e) {
            log.warn("[TRACKERS] SPAN 信息持久化失败: {}", e.getMessage());
            return false;
        }
    }


    public static void main(String[] args) {
        TrackerMetricServiceByElasticSearch service = new TrackerMetricServiceByElasticSearch();
        List<SpanNode> spans = new ArrayList<>();

        SpanNode span1 = new SpanNode();
        span1.setAppName("节点1");
        span1.setSpanName("节点1");
        span1.setSpanType("REDIS");
        span1.setSpanParentId("ROOT_SPAN");
        span1.setSpanId("1");
        span1.setTraceId("traceId");
        span1.setSpanStart(1);
        span1.setSpanEnd(2);
        spans.add(span1);

//        SpanNode span2 = new SpanNode();
//        span2.setAppName("节点2");
//        span2.setSpanName("节点2");
//        span2.setSpanType("REDIS");
//        span2.setSpanParentId("1");
//        span2.setSpanId("2");
//        span2.setTraceId("traceId");
//        span2.setSpanStart(2);
//        span2.setSpanEnd(3);
//        spans.add(span2);

        SpanNode span3 = new SpanNode();
        span3.setAppName("节点3");
        span3.setSpanName("节点3");
        span3.setSpanType("REDIS");
        span3.setSpanParentId("2");
        span3.setSpanId("3");
        span3.setTraceId("traceId");
        span3.setSpanStart(3);
        span3.setSpanEnd(4);
        spans.add(span3);

        List<SpanNode> sortSpan = service.sortTreeSpans(service.fillParentWithOrphan(spans));
//        List<SpanNode> sortSpan = controller.fillParentWithOrphan(spans);
        for (SpanNode spanNode : sortSpan) {
            System.out.println(spanNode.toString());
        }
    }

}
